package 多线程练习.Restaurant;

import java.util.concurrent.*;
import java.util.Random;

@FunctionalInterface
interface TaskProcessor<T> {
    void process(T task) throws InterruptedException;
}

class Task<T> {
    private final T data;

    public Task(T data) {
        this.data = data;
    }

    public T getData() {
        return data;
    }
}

class ProcessorFactory {
    public static TaskProcessor<String> createStringProcessor() {
        return task -> {
            Thread.sleep(80);
            System.out.println(Thread.currentThread().getName() + " 处理STRING: " + task.toUpperCase());
        };
    }

    public static TaskProcessor<Integer> createNumberProcessor() {
        return task -> {
            Thread.sleep(100);
            int result = task * task;
            System.out.println(Thread.currentThread().getName() + " 处理NUMBER: " + task + "²=" + result);
        };
    }
}

public class test3 {
    private static final int QUEUE_CAPACITY = 10;
    private final BlockingQueue<Task<?>> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);
    private final ExecutorService executor = Executors.newFixedThreadPool(5);
    private volatile boolean running = true;

    public void start() {
        // 启动3个生产者
        for (int i = 1; i <= 3; i++) {
            final int pid = i;
            executor.submit(() -> {
                try {
                    produceTasks(pid);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        // 启动2个消费者
        for (int i = 0; i < 2; i++) {
            executor.submit(() -> {
                try {
                    consumeTasks();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
    }

    private void produceTasks(int producerId) throws InterruptedException {
        Random rand = new Random();
        String[] samples = {"apple", "banana", "cherry", "END"};

        for (String sample : samples) {
            Task<?> task;
            if ("END".equals(sample) && producerId == 3) {
                task = new Task<>(sample);
            } else {
                task = (producerId % 2 == 0) ?
                        new Task<>(rand.nextInt(50)) :
                        new Task<>(sample + "-" + producerId);
            }

            queue.put(task);
            System.out.printf("Producer-%d 添加: %s%n", producerId, task.getData());
            Thread.sleep(rand.nextInt(200));
        }
    }

    private void consumeTasks() throws InterruptedException {
        TaskProcessor<String> stringProcessor = ProcessorFactory.createStringProcessor();
        TaskProcessor<Integer> numberProcessor = ProcessorFactory.createNumberProcessor();

        while (running && !Thread.currentThread().isInterrupted()) {
            Task<?> task = queue.take();
            Object data = task.getData();

            if ("END".equals(data)) {
                shutdown();
                return;
            }

            if (data instanceof String) {
                stringProcessor.process((String) data);
            } else if (data instanceof Integer) {
                numberProcessor.process((Integer) data);
            }
        }
    }

    public void shutdown() {
        running = false;
        executor.shutdownNow();
        try {
            if (!executor.awaitTermination(800, TimeUnit.MILLISECONDS)) {
                System.err.println("强制关闭剩余线程");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("系统已关闭");
    }

    public static void main(String[] args) {
        test3 system = new test3();
        system.start();
    }
}
